package com.example.media.bloom;

import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;

@Component
@RequiredArgsConstructor(onConstructor_ = @Autowired)
public class RedisBloomFilter {

    private final RedisTemplate redisTemplate;

    /**
     * 根据给定的布隆过滤器添加值，在添加一个元素的时候使用，批量添加的性能差
     *
     * @param key   KEY
     * @param value 值
     */
    public void add(String key, Object value) {
        int[] offset = BloomFilterHelper.murmurHashOffset(value);
        Arrays.stream(offset).parallel().forEach(i -> {
            redisTemplate.opsForValue().setBit(key, i, true);
        });
    }

    /**
     * 根据给定的布隆过滤器添加值，在添加一批元素的时候使用，批量添加的性能好，使用pipeline方式(如果是集群下，请使用优化后RedisPipeline的操作)
     *
     * @param key       KEY
     * @param valueList 值，列表
     */
    public void addList(String key, List valueList) {
        redisTemplate.executePipelined(new RedisCallback() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                valueList.parallelStream().forEach(value -> {
                    int[] offset = BloomFilterHelper.murmurHashOffset(value);
                    Arrays.stream(offset).parallel().forEach(i -> {
                        connection.setBit(key.getBytes(), i, true);
                    });
                });
                return null;
            }
        });
    }

    /**
     * 根据给定的布隆过滤器判断值是否存在
     *
     * @param key   KEY
     * @param value 值
     * @return 是否存在
     */
    public boolean contains(String key, Object value) {
        int[] offset = BloomFilterHelper.murmurHashOffset(value);
        return Arrays.stream(offset).parallel().allMatch(ints -> {
            if (redisTemplate.opsForValue().getBit(key, ints)) {
                return true;
            } else {
                return false;
            }
        });
    }


}
